﻿using System.Collections.Generic;
using System;
using System.Diagnostics;
using System.Linq;
using System.Linq.Expressions;
using System.Net;
using System.Text;
using System.Threading;
using GPC_Overlay;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using Microsoft.WindowsAzure.Diagnostics;
using System.IO;

namespace GISWorkerRole
{
    public class WorkerRole : RoleEntryPoint
    {
        string connectionString;
        static string connectionstring;
        string instance_id;
        string baseLayer, overlayLayer, op;
        public static int concurrencyLimit = 8;
        public static int currentOps = 0;
        public static int processed = 0;
        public static int toprocess = 0;
        public static List<string> blobs;
        public static List<Polygon> basePoly;
        public static List<Polygon> overlayPoly;
        int numMessages;
        List<string> taskIds;
        
        TimeSpan fileDownloadTime = new TimeSpan(0, 0, 0); /* Time to download files from cloud storage */
        TimeSpan parseTime = new TimeSpan(0, 0, 0); /* time to download files and parse them into C# objects */

        TimeSpan graphCreation = new TimeSpan(0, 0, 0);/* Time to create intersection graph */

        TimeSpan totalTaskCreationTime = new TimeSpan(0, 0, 0); /* Time to convert each task to XML and writing it to Cloud storage */
        TimeSpan taskUploadingTime = new TimeSpan(0, 0, 0); /* Time to write tasks to cloud storage */

        TimeSpan taskDownloadingTime = new TimeSpan(0, 0, 0); /* Time to download tasks for processing from cloud */
        TimeSpan processingTime = new TimeSpan(0, 0, 0); /* time to download a task, convert to polygons, and process using third party library */
       

        public override void Run()
        {
            connectionString = RoleEnvironment.GetConfigurationSettingValue("DataConnectionString");
            connectionstring = connectionString;
            string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
            while (true)
            {
                //The instanceIndex of the first instance is 0. 
                //Response.Write(instanceIndex);
                fileDownloadTime = new TimeSpan(0, 0, 0); /* Time to download files from cloud storage */
                parseTime = new TimeSpan(0, 0, 0); /* time to download files and parse them into C# objects */

                graphCreation = new TimeSpan(0, 0, 0);/* Time to create intersection graph */

                totalTaskCreationTime = new TimeSpan(0, 0, 0); /* Time to convert each task to XML and writing it to Cloud storage */
                taskUploadingTime = new TimeSpan(0, 0, 0); /* Time to write tasks to cloud storage */

                taskDownloadingTime = new TimeSpan(0, 0, 0); /* Time to download tasks for processing from cloud */
                processingTime = new TimeSpan(0, 0, 0); /* time to download a task, convert to polygons, and process using third party library */
           
                

                QueueHelper QueueObject = new QueueHelper(connectionString);
                CloudQueueMessage msg = new CloudQueueMessage("");
                DateTime waitingStart = DateTime.Now;

                while (!QueueHelper.PeekMessage(connectionString, DataConstants.INPUT_PARAS, out msg)) { }

                taskIds = new List<string>();
                System.Diagnostics.Trace.WriteLine("Got the inputs and parsing starts at: " + DateTime.Now.ToString(), "Information");
                DateTime parsingStart = DateTime.Now;
                _parseFiles(msg);
                parseTime = parseTime.Add(DateTime.Now.Subtract(parsingStart));
                System.Diagnostics.Trace.WriteLine("Parsing ends and task creation starts at: " + DateTime.Now.ToString(), "Information");

                DateTime taskCreationStart = DateTime.Now;
                _createGraph();
                graphCreation = graphCreation.Add(DateTime.Now.Subtract(taskCreationStart));
                System.Diagnostics.Trace.WriteLine("task creation ends and blob uploading starts at: " + DateTime.Now.ToString(), "Information");

                DateTime writingStart = DateTime.Now;
                _writeTasks();
                totalTaskCreationTime = totalTaskCreationTime.Add(DateTime.Now.Subtract(writingStart));
                System.Diagnostics.Trace.WriteLine("Blob uploading ends and overlay processing starts at " + DateTime.Now.ToString(), "Information");

                DateTime processingStart = DateTime.Now;

                while (taskIds.Count > 0)
                {
                    if (_handleBlob(taskIds[0], DateTime.Now))
                    {
                        taskIds.RemoveAt(0);
                    }
                }

                processingTime = processingTime.Add(DateTime.Now.Subtract(processingStart));
                processed = 1;
              
                /*
                 * string result = "File Downloading Time:\r\n" + fileDownloadTime.TotalSeconds.ToString() + " Seconds\r\nTotal Parsing Time:" + 
                    parseTime.TotalSeconds.ToString() + " Seconds\r\nGraph Creation Time:" + graphCreation.TotalSeconds.ToString() + " Seconds\r\nTotal Task Storing time:" + 
                    totalTaskCreationTime.TotalSeconds.ToString() + " Seconds\r\n Task Uploading Time:" + taskUploadingTime.TotalSeconds.ToString() +
                    " Seconds\r\n Total Processing Time:" + processingTime.TotalSeconds.ToString() + " Seconds\r\n Task Downloading time:" + taskDownloadingTime.TotalSeconds.ToString() +
                    " Seconds\r\n Role ID:" + instanceId.ToString();
                 */

                QueueHelper.PutMessageStatic(connectionString, DataConstants.TIME_QUEUE, new CloudQueueMessage(
                    fileDownloadTime.TotalSeconds.ToString() + "{" + parseTime.TotalSeconds.ToString() + "{" + graphCreation.TotalSeconds.ToString() + "{" +
                    totalTaskCreationTime.TotalSeconds.ToString() + "{" + taskUploadingTime.TotalSeconds.ToString() + "{" + processingTime.TotalSeconds.ToString() + "{" +
                    taskDownloadingTime.TotalSeconds.ToString() + "{"
                ));

                System.Diagnostics.Trace.WriteLine("Processing ends at: " + DateTime.Now.ToString(), "Information");
                while (QueueHelper.PeekMessage(connectionString, DataConstants.INPUT_PARAS, out msg)) ;
            }
            /*catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine("Error in worker role" + e.Message, "Information");
            }*/
        }

       
        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 48;

            this.instance_id = RoleEnvironment.CurrentRoleInstance.Id.ToString();
            
            // For information ons handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.
            CloudStorageAccount.SetConfigurationSettingPublisher((configName, configSetter) =>
            {
                // Provide the configSetter with the initial value
                configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));

                RoleEnvironment.Changed += (sender, arg) =>
                {
                    foreach (var change in arg.Changes)
                    {
                        var configChange = change as RoleEnvironmentConfigurationSettingChange;
                        if (configChange != null)
                        {
                            RoleEnvironment.GetConfigurationSettingValue(
                                configChange.ConfigurationSettingName);
                        }
                    }
                };
            });


            var diag = DiagnosticMonitor.GetDefaultInitialConfiguration();
            diag.Logs.ScheduledTransferLogLevelFilter = LogLevel.Verbose;
            diag.Logs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.WindowsEventLog.DataSources.Add("Application!*");
            diag.WindowsEventLog.DataSources.Add("System!*");
            diag.WindowsEventLog.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.DiagnosticInfrastructureLogs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            DiagnosticMonitor.Start("Microsoft.WindowsAzure.Plugins.Diagnostics.ConnectionString", diag);

            return base.OnStart();
        }

        private void _readnames(CloudQueueMessage msg)
        {   
            string[] Names = msg.AsString.Split('[');
            baseLayer = Names[0];
            overlayLayer = Names[1];
            op = Names[2];
        }

        private void _parseFiles(CloudQueueMessage msg)
        {
            DateTime removeStart = DateTime.Now;
            Parser layer1 = null, layer2 = null;

            List<string> msgids = new List<string>();
            /* Find the id of current role instance */
            string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
            int roleID = 0;
            if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
            {
                int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
            }
            /* Create space in local hard disk of this instance */
            LocalResource localCache = RoleEnvironment.GetLocalResource("maps");
            string localCacheRootDirectory = localCache.RootPath;
            _readnames(msg);
            /* download source XML files */

            string mapFirst = Path.Combine(localCacheRootDirectory, "base" + roleID);
            string mapSecond = Path.Combine(localCacheRootDirectory, "overlay" + roleID);
            BlobHelper blobObject = new BlobHelper(connectionString);
            DateTime downloadtime = DateTime.Now;
            {
                blobObject.BlobStreamDownload(DataConstants.MAP_CONTAINER_BLOB, baseLayer, mapFirst);
                blobObject.BlobStreamDownload(DataConstants.MAP_CONTAINER_BLOB, overlayLayer, mapSecond);
            }
            fileDownloadTime = fileDownloadTime.Add(DateTime.Now.Subtract(downloadtime));
          
            /* Parse GML files */
            layer1 = new Parser(mapFirst);
            layer2 = new Parser(mapSecond);

           
            numMessages = layer1.poly.Count;

           

            QueueHelper.PutMessageStatic(connectionString, DataConstants.BLOB_COUNT, new CloudQueueMessage(layer1.poly.Count.ToString()));

            /* Leave only local work in basePolygons */
            try
            {
                int fromPoly = 0, toPoly = 0;
                int RoleCount = RoleEnvironment.CurrentRoleInstance.Role.Instances.Count();
                int remainder = layer1.poly.Count % RoleCount;
                fromPoly = roleID * ((int)Math.Floor((double)layer1.poly.Count / RoleCount)) + Math.Min(roleID, remainder);
                toPoly = (roleID + 1) * ((int)Math.Floor((double)layer1.poly.Count / RoleCount)) + Math.Min(roleID + 1, remainder) - 1;

                layer1.poly.RemoveRange(toPoly + 1, layer1.poly.Count - toPoly - 1);
                layer1.poly.RemoveRange(0, fromPoly);
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine("Error finding Index - " + e.Message, "Information");
            }
            basePoly = new List<Polygon>();
            overlayPoly = new List<Polygon>();
           
           
            basePoly = layer1.poly;
            overlayPoly = layer2.poly;
           

            layer1.poly = null;
            layer2.poly = null;
           
        }

        private void _createGraph()
        {
            /* Create the graph of relationships between web role and worker roles */
            ParallelSort.QuicksortParallel(overlayPoly, true);
            Parser.createRelationships(ref basePoly,ref overlayPoly);
            overlayPoly.Clear();
        }

        private void _writeTasks()
        {
            /* Put independent tasks into blob storage, one blob per task */
            blobs = new List<string>();
            StringBuilder task = new StringBuilder();

            foreach (Polygon taskBlob in basePoly)
            {
                task.Clear();
                task.Append(Parser.PolygonToString(taskBlob));
                try
                {
                    foreach (Polygon overlayPolygon in taskBlob.overlayPolygons)
                    {
                        string polygonasstring = Parser.PolygonToString(overlayPolygon);
                        if (polygonasstring.Length > 0)
                        {
                            task.Append("[" + polygonasstring);
                        }
                    }
                    string connectionString = RoleEnvironment.GetConfigurationSettingValue("DataConnectionString");
                    string id = Guid.NewGuid().ToString();
                    DateTime startUpload = DateTime.Now;
                    BlobHelper.PutBlobText(connectionString, DataConstants.OUTPUT_CONTAINER_BLOB, id, task.ToString());
                    taskUploadingTime = taskUploadingTime.Add(DateTime.Now - startUpload);
                    taskIds.Add(id);
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.WriteLine("Error processing " + e.Message, "Information");
                }
            }
        }
       
        private bool _handleBlob(string blobid, DateTime timeToRead)
        {
            BlobHelper blobObject = new BlobHelper(connectionString);
            string polygonMsg = "";
            DateTime startDownload = DateTime.Now;
            try
            {
                blobObject.GetBlob(DataConstants.OUTPUT_CONTAINER_BLOB, blobid, out polygonMsg);
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine(e.Message + " :Error during GetBlob.");
                return false;
            }

            taskDownloadingTime = taskDownloadingTime.Add(DateTime.Now.Subtract(startDownload));

            /* The blob contains the GMl. We need to convert it to Polygon Object */
            Polygon blob = new Polygon();
            try
            {
                blob = Parser.PolygonFromString(polygonMsg);
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine(e.Message + " :Error during Polygon From String - worker role. PolygonMsg = " + polygonMsg);
                return false;
            }

            /* Since we have the task let us process it. Yippee! */
            try
            {
                _processPolygon(blob, blobObject, blobid);
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine(e.Message + " :Error during process Polygon.");
                return false;
            }

            /* If everything went fine, God forbid, let us get rid of the corpse */
            try
            {
                blobObject.DeleteBlob(DataConstants.OUTPUT_CONTAINER_BLOB, blobid);
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine(e.Message + " :Error during residue deletion.");
                return false;
            }
            return true;
        }

        private void _processPolygon(Polygon polygon, BlobHelper blobObject, string blobID)
        {

            QueueHelper queueObject = new QueueHelper(connectionString);
            Polygon rslt = new Polygon();
            StringBuilder resultString = new StringBuilder();

            int count = 0;

            try
            {
                /* find how many clipping operations are to be done and do them in a loop */
                count = polygon.overlayPolygons.Count;
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine("Error in counting overlay polygon" + e.Message);
            }
            /* Obviously this can be parallelized but we only use a single core worker */

   
            for (int r = 0; r < count; r++)
            {
                /* Release the kraken and let it take care of polygons */
                try
                {
                    rslt = polygon.Clip((GpcOperation)Enum.Parse(typeof(GpcOperation), op), polygon.overlayPolygons[r]);
                    /* Clipper library doesn't store params */
                    rslt.fID = polygon.fID + "-" + polygon.overlayPolygons[r].fID;
                    rslt.srsname = polygon.srsname;
                    rslt.srsDimension = polygon.srsDimension;
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.WriteLine(e.Message + " :Error during GPC Clipping.");
                }

                /* The clipping library returned the resultant polygon. Let us save it to our favorite GML format. */
                try
                {
                    resultString.Append(Parser.PolygonToString(rslt)); /* <-- This line is the main kicker. */
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.WriteLine(e.Message + " Error after clipping" + Parser.PolygonToString(rslt));
                }
            }
          
            try
            {
                string blockid = Guid.NewGuid().ToString();
                if (resultString.ToString().Length > 0)
                {
                    /* So private, you have the resultant GML. You are on cloud 9, can your result not be on cloud? Put me there sir!*/
                    BlobHelper.PutBlock(connectionString, DataConstants.MAP_CONTAINER_BLOB, DataConstants.OUTPUT_BLOB, blockid, resultString.ToString());
                }
               
                /* Web role is getting paid without doing anything. Let us store a message so that web role can keep track of how many messages we actually processed. 
                    * This is for book keeping, needed for reporting graphs in reports. It is not doing anything good to the process. However, if you need a notification of things ending you need this.*/
                ThreadPool.QueueUserWorkItem(putMessage, "done");
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine(e.Message + " :Error during put block/message - " + e.InnerException);
            }
        }
        private static void putMessage(object msg)
        {
            QueueHelper.PutMessageStatic(connectionstring , DataConstants.RESULT_QUEUE, new CloudQueueMessage((string) msg));
        }
    }
}
